package com.glodon.paas.qto.service.impl;

import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Calendar;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.StringTokenizer;

import javax.annotation.PostConstruct;

import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import com.glodon.paas.exception.PaasServerExceptionFactory;
import com.glodon.paas.job.JobConstants;
import com.glodon.paas.job.model.TaskStatus;
import com.glodon.paas.qto.QtoConstants;
import com.glodon.paas.qto.dao.JobRepository;
import com.glodon.paas.qto.dao.TaskRepository;
import com.glodon.paas.qto.model.Job;
import com.glodon.paas.qto.model.Task;
import com.glodon.paas.qto.model.TaskProgress;
import com.glodon.paas.qto.service.DispatcherService;
import com.glodon.paas.qto.service.QtoService;

@Service
public class QtoServiceImpl implements QtoService {
    private final Logger logger = LoggerFactory.getLogger(getClass());
    private final String BASE = "0123456789abcdefghigklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ";
    private final int COUNT = 6;
    private final DateFormat dateFormatter = new SimpleDateFormat("yyyy-MM-dd");

    @Value("${qto.query.span:1}")
    private int jobSpan;
    @Value("${job.task.workCount:3}")
    private int workCount;
    @Value("${qto.input.suffix:.gcl10}")
    private String inputSuffix;
    @Value("${qto.task.output.url}")
    private String taskOutputUrl;
    @Value("${qto.ip.converter}")
    private String convertIp;
    private final Map<String, List<String>> ipPair = new LinkedHashMap<String, List<String>>();

    @Autowired
    private JobRepository jobRepo;
    @Autowired
    private TaskRepository taskRepo;
    @Autowired
    private DispatcherService dispatcherService;

    @SuppressWarnings({ "rawtypes", "unchecked" })
    @Override
    public String create(Job job) {
        List<String> stories = splitStories(job.getStories(), workCount);
        
        jobRepo.save(job);
        logger.info("created a job: {}", job);
        // 创建虚拟父Task
        Task task = new Task(QtoConstants.QTO_CALC_TASK, job.getId(), null, null, null, true, stories.size());
        String outputPath = taskOutputUrl.trim() + "/" + dateFormatter.format(new Date()) + "/" + job.getId();
        task.setOutput(outputPath);
        taskRepo.save(task);
        logger.info("created a virtual task: {}", task);
        
        String fileName = generateKey();
        for (int i = 0; i < stories.size(); i++) {
            // 创建子Task
            Task childTask = new Task(QtoConstants.QTO_CALC_TASK, job.getId(), task.getId(), convertIpAddress(job.getInput()),
                    stories.get(i), false, 0);
            childTask.setStatus(TaskStatus.created);
            childTask.setOutput(outputPath);
            taskRepo.save(childTask);
            logger.debug("created a task: {}", childTask);
            Map params = new HashMap();
            Map inputMap = new HashMap();
            inputMap.put(fileName, appendUrlParam(childTask.getInput(), "userId=" + job.getUserId()));
            params.put(JobConstants.INPUT, inputMap);
            params.put(JobConstants.STORIES, StringUtils.split(childTask.getStories(), ","));
            params.put(JobConstants.OUTPUT, appendUrlParam(childTask.getOutput(), "userId=" + job.getUserId()));
            params.put(JobConstants.JOB_ID, childTask.getJobId());
            // params.put(JobConstants.USER_ID, job.getUserId());
            String jobServerTaskId = dispatcherService.dispath(QtoConstants.QTO_CALC_TASK, params);
            childTask.setJobServerTaskId(jobServerTaskId);
            logger.info("dispatched the task to job server: {}", childTask);
            taskRepo.save(childTask);
        }
        return job.getId();
    }

    @Override
    @Transactional
    public boolean processTaskNotify(String jobServerTaskId, TaskProgress progress) {
        taskRepo.modifyTaskProgress(jobServerTaskId, progress);
        boolean jobComplete = false;
        List<Task> taskList = taskRepo.findByJobServerTaskId(jobServerTaskId);
        if (taskList == null || taskList.isEmpty()) {
            throw PaasServerExceptionFactory.notFoundException("invalid taskId:" + jobServerTaskId, null);
        }
        Task task = taskList.get(0);
        if (TaskStatus.success.equals(progress.getStatus())) {// Task已经完成
            String parentId = task.getParentId();
            // 2. 判断是否存在父任务
            if (parentId == null) {// 不存在父任务,job已全部完成
                jobComplete = true;
            } else {// 存在父任务
                Task parentTask = taskRepo.findOne(parentId);
                // 2.1 判断父任务是否是虚拟任务
                if (parentTask.isVirtualTask()) {// 虚拟任务
                    // 检查是否为最后一个反馈的子任务
                    boolean calcComplete = isCalcluateTaskAllComplete(parentTask);
                    if (calcComplete) {// 最后一个子任务，取下一任务提交.
                        boolean success = taskRepo.modifyVirtualTask(parentId);
                        if (success) {
                            dispatchMergeTask(parentTask);
                        }
                    }
                }
            }
            if (jobComplete == true) {// 更新Job完成时间戳
                Job job = jobRepo.findOne(task.getJobId());
                job.setEndTime(new Date());
                // job.setOutput(progress.getOutput());
                job.setFilename(progress.getFilename());
                jobRepo.save(job);
            }
        }
        return jobComplete;
    }

    @Override
    public Job getJob(String jobId) {
        Job job = jobRepo.findOne(jobId);
        if (job == null) {
            throw PaasServerExceptionFactory.notFoundException("invalid jobId:" + jobId, null);
        }
        return job;
    }

    @Override
    public List<Job> findByUserId(String userId, Date fromTime, Date toTime) {
        if (toTime == null) {
            toTime = new Date();
        }
        if (fromTime == null) {
            Calendar calendar = Calendar.getInstance();
            calendar.setTime(toTime);
            calendar.set(Calendar.DATE, -1 * jobSpan);
            fromTime = calendar.getTime();
        }
        return jobRepo.findByUserIdAndStartTimeBetweenOrderByStartTimeDesc(userId, fromTime, toTime);
    }

    @Override
    public List<Task> findChildTask(String jobId) {
        return taskRepo.findByJobIdAndVirtualTaskIsFalseOrderByStartTimeAsc(jobId);
    }
    
    @PostConstruct
    public void initConvertIp() {
        if (StringUtils.isNotEmpty(convertIp)) {
            String[] group = StringUtils.split(convertIp, ";");
            for (int i = 0; i < group.length; i++) {
                String[] pairs = StringUtils.split(group[i], "|");
                String[] ipArray = StringUtils.split(pairs[1], ",");
                List<String> ips = Arrays.asList(ipArray);
                ipPair.put(pairs[0], ips);
            }
        }
    }
    
    /**
     * 将url中的外部ip地址替换为内网ip地址，提高agent文件传输速度.
     * 
     * @param url
     * @return
     */
    private String convertIpAddress(String url) {
        for (String externalIP : ipPair.keySet()) {
            if (url.contains(externalIP)) {
                List<String> internalIPs = ipPair.get(externalIP);
                String internalUrl = StringUtils.replaceOnce(url, externalIP, internalIPs.get(0));
                return internalUrl;
            }
        }
        return url;
    }

    @SuppressWarnings({ "rawtypes", "unchecked" })
    private boolean dispatchMergeTask(Task parentTask) {
        Job job = jobRepo.findOne(parentTask.getJobId());
        Task mergeTask = new Task(QtoConstants.QTO_MERGE_TASK, parentTask.getJobId(), null, null, null, false, 0);
        mergeTask.setStatus(TaskStatus.created);
        mergeTask.setInput(parentTask.getOutput());
        mergeTask.setOutput(job.getOutput());
        Map params = new HashMap();
        Map inputMap = new HashMap();
        inputMap.put(System.currentTimeMillis(), appendUrlParam(parentTask.getOutput(), "userId=" + job.getUserId()));

        String outputUrl = convertIpAddress(job.getOutput());
        outputUrl = appendUrlParam(outputUrl, "userId=" + job.getUserId());
        taskRepo.save(mergeTask);
        params.put(JobConstants.INPUT, inputMap);
        params.put(JobConstants.OUTPUT, outputUrl);
        params.put(JobConstants.JOB_ID, parentTask.getJobId());
        // params.put(JobConstants.USER_ID, job.getUserId());
        params.put(JobConstants.PROPERTY_TASK_DELETE_INPUT, true);
        String taskId = dispatcherService.dispath(QtoConstants.QTO_MERGE_TASK, params);
        mergeTask.setJobServerTaskId(taskId);
        taskRepo.save(mergeTask);
        return true;
    }

    /**
     * 所有的计算任务是否都已经结束了.
     * 
     * @param parentId
     *            虚拟的父任务Task Id
     * @return
     */
    private boolean isCalcluateTaskAllComplete(Task parentTask) {
        long completeCount = taskRepo.getChildTaskCompleteCount(parentTask.getId());
        if (completeCount == parentTask.getStoryCount()) {
            return true;
        } else {
            return false;
        }
    }

    /**
     * 根据楼层数和work数拆分楼层.
     * 
     * @param stories
     * @param workCount
     * @return
     */
    private List<String> splitStories(String stories, int workCount) {
        List<String> list = null;
        if (stories == null || stories.isEmpty()) {
            return Collections.emptyList();
        } else {
            StringTokenizer tokenizer = new StringTokenizer(stories, ",");
            int storyCount = tokenizer.countTokens();
            int averageCount = storyCount / workCount;
            int remainder = storyCount % workCount;
            int group = averageCount != 0 ? workCount : remainder;
            list = new ArrayList<String>(group);
            for (int i = 0; i < group; i++) {
                int workPayload = averageCount;
                if (i >= group - remainder) {
                    workPayload += 1;
                }
                StringBuffer sb = new StringBuffer();
                for (int j = 0; j < workPayload; j++) {
                    sb.append(tokenizer.nextToken()).append(",");
                }
                list.add(sb.substring(0, sb.length() - 1));
            }
        }
        return list;
    }

    /**
     * 在url后面追加一个参数.
     * 
     * @param url
     * @param keyValueParam
     * @return
     */
    private String appendUrlParam(String url, String keyValueParam) {
        if (url.indexOf("?") < 0) {
            return url + "?" + keyValueParam;
        } else if (url.endsWith("?")) {
            return url + keyValueParam;
        } else {
            return url + "&" + keyValueParam;
        }
    }

    private String generateKey() {
        StringBuffer sb = new StringBuffer();
        Random random = new Random(System.currentTimeMillis());
        for (int i = 0; i < COUNT; i++) {
            int nextInt = random.nextInt(62);
            sb.append(BASE.charAt(nextInt));
        }
        sb.append(inputSuffix);
        return sb.toString();
    }

    public void setDispatcherService(DispatcherService dispatcherService) {
        this.dispatcherService = dispatcherService;
    }

    public void setTaskRepo(TaskRepository taskRepo) {
        this.taskRepo = taskRepo;
    }

    public void setJobRepo(JobRepository jobRepo) {
        this.jobRepo = jobRepo;
    }

    public void setJobSpan(int jobSpan) {
        this.jobSpan = jobSpan;
    }

    public void setWorkCount(int workCount) {
        this.workCount = workCount;
    }
    public void setConvertIp(String convertIp) {
        this.convertIp = convertIp;
    }
}
